//package com.dfd.producer.task;
//
//import cn.hutool.core.collection.CollectionUtil;
//import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
//import com.dfd.common.mq.po.MqRecord;
//import com.dfd.producer.service.MqRecordService;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.producer.DefaultMQProducer;
//import org.apache.rocketmq.client.producer.SendResult;
//import org.apache.rocketmq.client.producer.SendStatus;
//import org.apache.rocketmq.common.message.Message;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.scheduling.annotation.Scheduled;
//import org.springframework.stereotype.Component;
//import org.springframework.util.CollectionUtils;
//
//import java.util.List;
//import java.util.stream.Collectors;
//
///**
// * 处理未发送成功的消息
// *
// * @Author: Fengdong.Duan
// * @Date: 2023/2/9 9:08
// */
//@Component
//@Slf4j
//public class MqTask {
//    @Autowired
//    MqRecordService mqRecordService;
//    @Autowired
//    DefaultMQProducer defaultMQProducer;
//
//    @Scheduled(cron = "0 0/1 * * * ?")
//    public void mqTask() {
//        List<MqRecord> notAvailable = mqRecordService.list(new LambdaQueryWrapper<MqRecord>()
//                .ne(MqRecord::getConsumerStatus, true));
//        if (CollectionUtils.isEmpty(notAvailable)) {
//            log.info("没有消息需要重新发送，(￣▽￣)~*");
//            return;
//        }
//        List<Message> messages = notAvailable.stream().map(mqRecord -> {
//            Message message = new Message();
//            message.setKeys(String.valueOf(mqRecord.getBusinessId()));
//            message.setTopic(mqRecord.getTopic());
//            message.setTags(mqRecord.getTag());
//            message.setBody(mqRecord.getContent().getBytes());
//            return message;
//        }).collect(Collectors.toList());
//        List<List<Message>> split = CollectionUtil.split(messages, 100);
//        try {
//            for (List<Message> messageList : split) {
//                SendResult send = defaultMQProducer.send(messageList);
//                if (SendStatus.SEND_OK.equals(send.getSendStatus())) {
//                    log.info("消息批量发送成功");
//                }
//            }
//        } catch (Exception e) {
//            log.error("消息发送失败：{}", e.getMessage());
//            e.printStackTrace();
//        }
//    }
//}
